package com.atguigu.flink.chapter05.transform.partition;


import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.MathUtils;

import javax.swing.tree.DefaultMutableTreeNode;

/*
keyBy:
    按照key进行分区（分组）
    相同key的元素，肯定在同一个分区
    一个分区可以有多个组
    key,128,2
        KeyGroupRangeAssignment.assignKeyToParallelOperator(
                key, maxParallelism, numberOfChannels);
                // 128, 2, [0,127]
                computeOperatorIndexForKeyGroup(
                maxParallelism, parallelism, assignToKeyGroup(key, maxParallelism));
                    // key, 128
                    assignToKeyGroup(key, maxParallelism)
                        // key 的hash 值, 128
                        computeKeyGroupForKeyHash(key.hashCode(), maxParallelism);
                            MathUtils.murmurHash(keyHash) % maxParallelism;
     [0,127] * 2 / 128
     keyGroupId * parallelism / maxParallelism;

       keyGroupId = MathUtils.murmurHash(keyHash) % maxParallelism

     1、对key 进行双重hash : hashCode->murmur hash
     2、KeyGroupId * parallelism / maxParallelism : 计算并行度


shuffle:
    随机分区

rebalance:
    平均分布

    跨TaskManager
rescale:
    平均分布
     不跨TaskManager
broadCast:
    广播

    一个元素会被广播到所有的分区
global:
    所有的元素进入0并行度
 */
public class PartitionDemo {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port",2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(2);

        env
                //.socketTextStream("hadoop102",8888)   socket的并行度永远是1
                .fromElements(10,21,31,41,51,61,90)
                //.keyBy(x -> x % 2)
                //.shuffle()   //随机分区
                //.rebalance()   //平均分布
                //.rescale()
                //.broadcast()
                //.global()
                .partitionCustom(new MyPartitioner(), new KeySelector<Integer, String>() {
                    @Override
                    public String getKey(Integer value) throws Exception {
                        return value + " ";
                    }
                })
                .print();

        //System.out.println(MathUtils.murmurHash("奇数".hashCode()) % 128);
        //System.out.println(MathUtils.murmurHash("偶数".hashCode()) % 128);
        //System.out.println(MathUtils.murmurHash(Integer.valueOf(0).hashCode()) % 128);
        //System.out.println(MathUtils.murmurHash(Integer.valueOf(1).hashCode()) % 128);

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}




/*
keyBy:
    按照key进行分区【分组】
    相同key的元素，肯定在同一个分区
    一个分区可以有多个组
    KeyGroupId = MathUtils.murmurhash(Keyhash) % maxParallelism

   1、对key进行双重hash: code hash -> murmurhash
   2、KeyGroupId * Parallelism / maxParamllelism : 计算并行度

shuffle:
    随机分区

rebalance:
    平均分布
    跨TaskManager

rescale:
    平均分布
    不跨TaskManager

broadCast:
    广播
    一个元素会被广播到所有分区

global:
   所有的元素会进入0并行度
 */

//public class PartitionDemo {
//    public static void main(String[] args) {
//        Configuration conf = new Configuration();
//        conf.setInteger("rest.port",2000);
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
//        env.setParallelism(2);
//
//        env.fromElements(10,21,31,40,51,61,90)
//                //.keyBy(x -> x % 2)
//                //.shuffle()
//                //.rebalance()
//                //.rescale()
//                //.broadcast()
//                //.global()
//                .partitionCustom(new MyPartitioner(), new KeySelector<Integer, String>() {
//                    @Override
//                    public String getKey(Integer value) throws Exception {
//                        return value + "";
//                    }
//                })
//                .print();
//
//        //System.out.println(MathUtils.murmurHash("奇数".hashCode()) % 128);
//        //System.out.println(MathUtils.murmurHash("偶数".hashCode()) % 128);
//        //System.out.println(MathUtils.murmurHash(Integer.valueOf(0).hashCode()) % 128);
//        //System.out.println(MathUtils.murmurHash(Integer.valueOf(1).hashCode()) % 128);
//        try {
//            env.execute();
//        } catch (Exception e) {
//            e.printStackTrace();
//        }
//    }
//}